#pragma once
#include "requestor.hpp"
#include <unordered_set>

/*
进行消息发布，消息订阅的客户端模块
*/

namespace bitrpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            using SubCallback = std::function<void(const std::string &key, const std::string &msg)>;
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager(const Requestor::ptr &requestor) : _requestor(requestor) {}
            bool create(const BaseConnection::ptr &conn, const std::string &key)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_CREATE);
            }
            bool remove(const BaseConnection::ptr &conn, const std::string &key)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_REMOVE);
            }
            bool subscribe(const BaseConnection::ptr &conn, const std::string &key, const SubCallback &cb)
            {
                /* 在订阅操作时，要把subcallback和std::string的映射关系建立起来，表示的是当消息客户端
                   进行消息推送的时候，服务器发来的request，不同的主题有不同的回调接口 */
                addSubscribe(key, cb);
                bool ret = commonRequest(conn, key, TopicOptype::TOPIC_SUBSCRIBE);
                if (ret == false)
                {
                    delSubscribe(key);
                    return false;
                }
                /* 订阅主题后，不进行消息的等待，等到消息来了，调用设置的回调函数进行异步处理 */
                return true;
            }
            bool cancel(const BaseConnection::ptr &conn, const std::string &key)
            {
                delSubscribe(key);
                return commonRequest(conn, key, TopicOptype::TOPIC_CANCEL);
            }
            bool publish(const BaseConnection::ptr &conn, const std::string &key, const std::string &msg)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_PUBLISH, msg);
            }
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            { /* 提供给dispatcher模块，收到服务器推送的消息 */
                // 1. 从消息中取出操作类型进行判断，是否是消息请求
                auto type = msg->optype();
                if (type != TopicOptype::TOPIC_PUBLISH)
                {
                    ELOG("收到了错误类型的主题操作！");
                    return;
                }
                // 2. 取出消息主题名称，以及消息内容
                std::string topic_key = msg->topicKey();
                std::string topic_msg = msg->topicMsg();
                // 3. 通过主题名称，查找对应主题的回调处理函数，有在处理，无在报错    ()
                auto callback = getSubscribe(topic_key);
                if (!callback)
                {
                    ELOG("收到了 %s 主题消息，但是该消息无主题处理回调！", topic_key.c_str());
                    return;
                }
                return callback(topic_key, topic_msg);
            }

        private:
            void addSubscribe(const std::string &key, const SubCallback &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert(std::make_pair(key, cb));
            }
            void delSubscribe(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }
            const SubCallback getSubscribe(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callbacks.find(key);
                if (it == _topic_callbacks.end())
                {
                    return SubCallback();
                }
                return it->second;
            }
            bool commonRequest(const BaseConnection::ptr &conn, const std::string &key,
                               TopicOptype type, const std::string &msg = "")
            { /* message作为Json_Value的一个字段存储msg_req->setTopicMsg(msg);  */
                // 1. 构造请求对象，并填充数据
                auto msg_req = MessageFactory::create<TopicRequest>();
                msg_req->setId(UUID::uuid());
                msg_req->setMType(MType::REQ_TOPIC);
                msg_req->setOptype(type);
                msg_req->setTopicKey(key);
                if (type == TopicOptype::TOPIC_PUBLISH)
                {
                    msg_req->setTopicMsg(msg);
                }
                // 2. 向服务端发送请求，等待响应
                BaseMessage::ptr msg_rsp;
                bool ret = _requestor->send(conn, msg_req, msg_rsp);
                if (ret == false)
                {
                    ELOG("主题操作请求失败！");
                    return false;
                }
                // 3. 判断请求处理是否成功
                auto topic_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(msg_rsp);
                if (!topic_rsp_msg)
                {
                    ELOG("主题操作响应，向下类型转换失败！");
                    return false;
                }
                if (topic_rsp_msg->rcode() != RCode::RCODE_OK)
                {
                    ELOG("主题操作请求出错：%s", errReason(topic_rsp_msg->rcode()).c_str());
                    return false;
                }
                return true;
            }

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, SubCallback> _topic_callbacks; /* dispatcher将推送消息分发到topic模块后，针对不同的主题有不同的回调 */
            Requestor::ptr _requestor;
        };
    }
}

/*
 这是rpc  客户端主题模块，负责主题的创建，删除，订阅，取消订阅以及推送消息
 需要提供一个接口给dispatcher，负责在收到推送信息的request后，根据不同的主题进行不同的回调处理这里的回调采用异步回调
*/
namespace bitrpc
{
    namespace client
    {
        class testTopicManager
        {
            testTopicManager(Requestor::ptr requstor) : _requstor(requstor) {}
            using ptr = std::shared_ptr<testTopicManager>();
            using SubCallback = std::function<void(std::string topic, std::string message)>;
            // 创建主题
            bool create(const BaseConnection::ptr &conn, const std::string topic)
            {
                return commonMessage(conn, topic, TopicOptype::TOPIC_CREATE);
            }

            bool remove(const BaseConnection::ptr &conn, const std::string topic)
            {
                return commonMessage(conn, topic, TopicOptype::TOPIC_CANCEL);
            }

            bool subscribe(const BaseConnection::ptr &conn, const std::string topic, SubCallback cb)
            {
                addSubscribe(topic,cb); //先要创建，不然等服务端给你推送消息了你还没有设置对应主题的回调函数
                bool ret = commonMessage(conn, topic, TopicOptype::TOPIC_SUBSCRIBE);
                if(ret == false)
                {
                    delSubscribe(topic);
                    return false;
                }
                return true;
            }

            bool cancel(const BaseConnection::ptr &conn, const std::string topic)
            {
                delSubscribe(topic);
                return commonMessage(conn, topic, TopicOptype::TOPIC_CANCEL);
            }

            bool publish(const BaseConnection::ptr &conn, const std::string topic,const std::string& msg)
            {
                return commonMessage(conn, topic, TopicOptype::TOPIC_PUBLISH,msg);
            }

            // 注册给dispatcher
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg) /* MType只是一个数字是Json的字段
                                                                              用来判断当前报文的类型来(创建对应的报文对象)，但本质不是个类型 */
            {
                auto type = msg->optype();
                if(type!= TopicOptype::TOPIC_PUBLISH)
                {
                    //意味着传过来的请求不是客户端的消息通知请求
                    ELOG("类型错误");
                    return ;
                }
                std::string topic_key = msg->topicKey();
                std::string topic_msg = msg->topicMsg();

                auto callback = getSubscribe(topic_key);
                if (!callback)
                {
                    ELOG("收到了 %s 主题消息，但是该消息无主题处理回调！", topic_key.c_str());
                    return;
                }
                return callback(topic_key, topic_msg);
    
            }

            void addSubscribe(const std::string &key, const SubCallback &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert(std::make_pair(key, cb));
            }
            void delSubscribe(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }
            const SubCallback getSubscribe(const std::string &key) /* 返回一个SubCallback对象 */
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callbacks.find(key);
                if (it == _topic_callbacks.end())
                {
                    return SubCallback();
                }
                return it->second;
            }

            bool commonMessage(const BaseConnection::ptr &conn, const std::string &key,
                               TopicOptype type, const std::string &msg = "")
            {
                TopicRequest::ptr mmsg = std::make_shared<TopicRequest>();
                mmsg->setId(UUID::uuid());
                mmsg->setMType(MType::REQ_TOPIC);
                mmsg->setOptype(type);
                mmsg->setTopicKey(key);
                if (type == TopicOptype::TOPIC_PUBLISH)
                {
                    mmsg->setTopicMsg(msg);
                }
                BaseMessage::ptr result;
                bool ret = _requstor->send(conn, mmsg, result); /// 进行同步调用，阻塞
                if (ret == false)
                {
                    ELOG("调用失败");
                    return false;
                }
                auto rsp_ptr = std::dynamic_pointer_cast<TopicResponse>(result);
                if (rsp_ptr == nullptr)
                {
                    ELOG("类型向下转换失败");
                    return false;
                }
                if ((rsp_ptr->rcode()) != RCode::RCODE_OK)
                {
                    ELOG("调用出错，错误码为: %s", errReason(rsp_ptr->rcode()).c_str());
                    return false;
                }
                return true;
            }

        private:
            std::mutex _mutex;
            BaseClient::ptr _client;
            Requestor::ptr _requstor;
            std::unordered_map<std::string, SubCallback> _topic_callbacks;
        };
    }
}